package com.cn;

import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class MsgExecutor extends ThreadPoolExecutor {
	private MsgQueuePool<Integer, AWork> pool = new MsgQueuePool<Integer, AWork>();

	private String name;
	private int corePoolSize;
	private int maxQueueSize;

	public MsgExecutor(String name, int corePoolSize, int maxQueueSize) {
		super(corePoolSize, 2 * corePoolSize, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
		this.name = name;
		this.corePoolSize = corePoolSize;
		this.maxQueueSize = maxQueueSize;
	}

	public MsgExecutor(int corePoolSize) {
		this("queue-pool", corePoolSize, 10000);
	}

	/**
	 * 增加执行任务
	 * @param key
	 * @param value
	 * @return
	 */
	public boolean addTask(Integer key, AWork task) {
		key = key % corePoolSize;
		TasksQueue<AWork> queue = pool.getTasksQueue(key);
		boolean run = false;
		boolean result = false;
		synchronized (queue) {
			if (maxQueueSize > 0) {
				if (queue.size() > maxQueueSize) {
					LogHelper.ERROR_LOGGER.error("队列" + name + "(" + key + ")" + "抛弃指令!");
					queue.clear();
				}
			}
			result = queue.add(task);
			if (result) {
				task.setTasksQueue(queue);
				{
					if (queue.isProcessingCompleted()) {
						queue.setProcessingCompleted(false);
						run = true;
					}
				}
			} else {
				LogHelper.ERROR_LOGGER.error("orderedqueue队列添加任务失败");
			}
		}
		if (run) {
			execute(queue.poll());
		}
		return result;
	}

	@Override
	protected void afterExecute(Runnable r, Throwable t) {
		super.afterExecute(r, t);
		AWork work = (AWork) r;
		TasksQueue<AWork> queue = work.getTasksQueue();
		if (queue != null) {
			AWork afterWork = null;
			synchronized (queue) {
				afterWork = queue.poll();
				if (afterWork == null) {
					queue.setProcessingCompleted(true);
				}
			}
			if (afterWork != null) {
				execute(afterWork);
			}
		}
	}

	/**
	 * 获取剩余任务数量
	 */
	public int getTaskCounts() {
		int count = super.getActiveCount();
		Iterator<Entry<Integer, TasksQueue<AWork>>> iter = pool.getTasksQueues().entrySet().iterator();
		while (iter.hasNext()) {
			Entry<Integer, TasksQueue<AWork>> entry = (Entry<Integer, TasksQueue<AWork>>) iter.next();
			TasksQueue<AWork> tasksQueue = entry.getValue();
			count += tasksQueue.size();
		}
		return count;
	}
}
